package com.example.dmweb.chat;

import com.alibaba.fastjson.JSON;
import com.example.dmcommom.*;
import com.example.dmdomain.*;
import com.example.dmdomain.entity.ChatContext;
import com.example.dmdomain.entity.ChatSession;
import com.example.dmdomain.entity.SystemChatContext;
import com.example.dmdomain.entity.SystemChatSession;
import com.example.dmservice.auth.AuthService;
import com.example.dmweb.chat.chatService.ChatModel;
import com.example.dmweb.chat.chatService.ChatRequest;
import com.example.dmweb.chat.chatService.EleBillModel;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import okhttp3.OkHttpClient;
import okhttp3.Request;
import okhttp3.Response;
import okhttp3.ResponseBody;
import okio.BufferedSource;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.http.MediaType;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import javax.annotation.Resource;
import javax.persistence.criteria.*;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.io.PrintWriter;
import java.time.Duration;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

@Slf4j
@RestController
@EnableScheduling
@Configuration
public class SSEController {
    @Resource
    ChatContextRepository contextRepository;
    @Resource
    SystemChatContextRepository systemChatContextRepository;
    @Resource
    ChatSessionRepository sessionRepository;
    @Resource
    SystemChatSessionRepository systemChatSessionRepository;
    @Resource
    private ChatFileRepository fileRepository;
    @Resource
    private SystemChatFileRepository systemChatFileRepository;

    @Value("${chat.server.elebill.url}")
    private String chatServer;

    @Value("${chat.server.authorization}")
    private String chatServerAuth;
    private final Map<String, SseEmitter> sessionEmitters = new ConcurrentHashMap<>();

    @PostMapping(value = "/chat", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    @NoAuthInterception
    public SseEmitter streamSse(HttpServletRequest request, @RequestBody ChatModel model,HttpServletResponse response) {
        response.addHeader("Cache-Control", "no-cache");
        response.addHeader("X-Accel-Buffering", "no");
        String useId = request.getHeader("Authorization");
        String sessionId = StringUtils.isEmpty(model.getSessionId()) ? UUID.randomUUID().toString() : model.getSessionId();
        log.info("sessionId: {}", sessionId);
        log.info("model传参:{}", JSON.toJSONString(model));
        Integer system = 0;
        StringBuilder fisContent = new StringBuilder();
        SseEmitter emitter = new SseEmitter(Long.MAX_VALUE);
        sessionEmitters.put(sessionId, emitter);

        AuthService service = SpringBeanHelper.getBeanOrThrow(AuthService.class);
        try {
            boolean auth = service.checkPermission(useId);
            if (!auth) {
                AuthException e = new AuthException("账户不存在或未授权!");
                handleError(emitter, null, e, system);
                return emitter;
            }
        } catch (Exception e) {
            log.error("auth error", e);
            handleError(emitter, null, e, system);
            return emitter;
        }

        ChatRequest chat = new ChatRequest(model);
        chat.setConversation_id(sessionId);
        chat.setInput(model.getInput());
        if (model.getSystem() == null || model.getSystem().isEmpty()) {
            system = 1;
            fileRepository.findAll((root, criteriaQuery, cb) -> {
                List<Predicate> predicates = new ArrayList<>();
                predicates.add(cb.in(root.get("fileId")).value(model.getFileId()));
                criteriaQuery.where(cb.and(predicates.toArray(new Predicate[0])));
                return criteriaQuery.getRestriction();
            }).forEach(v -> fisContent.append(v.getFileContent()));
        } else {
            chat.setSystem(model.getSystem());
            system = 2;
            systemChatFileRepository.findAll((root, criteriaQuery, cb) -> {
                List<Predicate> predicates = new ArrayList<>();
                predicates.add(cb.in(root.get("fileId")).value(model.getFileId()));
                criteriaQuery.where(cb.and(predicates.toArray(new Predicate[0])));
                return criteriaQuery.getRestriction();
            }).forEach(v -> fisContent.append(v.getFileContent()));
        }
        chat.setDoc_content(String.valueOf(fisContent));

        List<ChatSession> chatSession = sessionRepository.findByRequestId(model.getRequestId());
        if (chatSession != null && !chatSession.isEmpty()) {
            chatSession.forEach(v -> v.setDelState(1));
            sessionRepository.saveAll(chatSession);
        }
        List<SystemChatSession> systemChatSession = systemChatSessionRepository.findByRequestId(model.getRequestId());
        if (systemChatSession != null && !systemChatSession.isEmpty()) {
            systemChatSession.forEach(v -> v.setDelState(1));
            systemChatSessionRepository.saveAll(systemChatSession);
        }



        boolean[] isCompleted = {false};
        ChatContext context = new ChatContext();
        ChatSession session = new ChatSession();
        try {
            Date date = new Date();
            String contextId = sessionId + date.getTime();

            context.setId(contextId);
            context.setRequestId(model.getRequestId());
            context.setSessionTime(date);
            context.setSueContent(model.getInput());

            session.setId(contextId);
            session.setRequestId(model.getRequestId());
            session.setSessionTime(date);
            String truncatedInput = model.getInput().length() > 10 ? model.getInput().substring(0, 10) : model.getInput();
            session.setSessionName(truncatedInput);
            session.setUseId(useId);
            session.setDelState(0);
            if (model.getFileId() != null) {
                context.setFileIid(String.join(",", model.getFileId()));
                session.setFileId(String.join(",", model.getFileId()));
            }
            if (model.getSessionId() == null || model.getSessionId().isEmpty()) {
                session.setSessionId(sessionId);
                session.setDelState(3);
            }
            context.setSessionId(sessionId);
            session.setSessionId(sessionId);
            if (system == 1){
                sessionRepository.save(session);
            } else if (system == 2){
                systemChatSessionRepository.save(toSystem(session,model.getSystem()));
            }

            StringBuilder sysContent = new StringBuilder();

            sendSseEvent(emitter, new HashMap<String, String>() {{
                put("requestId", model.getRequestId());
                put("sessionId", model.getSessionId());
            }});

            OkHttpClient client = new OkHttpClient().newBuilder().readTimeout(Duration.ofMinutes(3)).build();
            okhttp3.MediaType mediaType = okhttp3.MediaType.get("application/json");
            okhttp3.RequestBody body = okhttp3.RequestBody.create(mediaType, JSON.toJSONString(chat));
            Request request11 = new Request.Builder()
                    .url(chatServer)
                    .header("Authorization", chatServerAuth)
                    .addHeader("Accept", "text/event-stream")
                    .post(body)
                    .build();
            Integer finalSystem = system;
            new Thread(() -> handleResponse(client, request11, emitter, isCompleted, sessionId, context, sysContent, finalSystem)).start();
        } catch (Exception e) {
            handleError(emitter, context, e,system);
        } finally {
            emitter.onCompletion(() -> {
                isCompleted[0] = true;
                emitter.complete();
                sessionEmitters.remove(sessionId);
                log.info("清除");
            });
            emitter.onTimeout(() -> {
                log.info("超时");
            });
        }
        return emitter;
    }

    private void handleResponse(OkHttpClient client, Request request, SseEmitter emitter, boolean[] isCompleted
            , String sessionId, ChatContext context, StringBuilder sysContent,Integer system) {
        try (Response response = client.newCall(request).execute()) {
            if (!response.isSuccessful())
                throw new IOException("Algorithm Request failed with HTTP error code: " + response.code());

            ResponseBody responseBody = response.body();
            if (responseBody == null || isCompleted[0]) return;
            BufferedSource source = responseBody.source();
            while (!source.exhausted()) {
                String data = source.readUtf8Line();
                if (data != null) {
                    ObjectMapper mapper = new ObjectMapper();
                    EleBillModel model = mapper.readValue(data, EleBillModel.class);
                    sysContent.append(model.getText());
                    context.setSysContent(String.valueOf(sysContent));
                    sendSseEvent(emitter, isCompleted, model, context,system);
                    if ("1".equals(model.getIs_finished())) {
                        if (system == 1){
                            contextRepository.save(context);
                        } else if (system == 2){
                            systemChatContextRepository.save(toSystem(context));
                        }
                        isCompleted[0] = true;
                        emitter.complete();
                        sessionEmitters.remove(sessionId);
                        break;
                    }
                }
            }
        } catch (IOException e) {
            sessionEmitters.remove(sessionId);
            handleError(emitter, context, e,system);
        } finally {
            if (system == 1){
                contextRepository.save(context);
            } else if (system == 2){
                systemChatContextRepository.save(toSystem(context));
            }
            log.info("储存");
        }
    }
    private void sendSseEvent(SseEmitter emitter, Object event) {
        synchronized (emitter) {
            try {
                emitter.send(SseEmitter.event().data(event));
            } catch (IOException e) {
                log.error("Error while sending SSE event first time: {}", e.getMessage());
                emitter.completeWithError(e);
            }
        }
    }

    private void sendSseEvent(SseEmitter emitter, boolean[] isCompleted, Object event, ChatContext context,Integer system) {
        synchronized (emitter) {
            if (isCompleted[0]) return;
            try {
                emitter.send(SseEmitter.event().data(event));
            } catch (IOException e) {
                if (system == 1){
                    contextRepository.save(context);
                } else if (system == 2){
                    systemChatContextRepository.save(toSystem(context));
                }
                log.error("Error while sending SSE event: {}", e.getMessage());
                emitter.completeWithError(e);
                isCompleted[0] = true;
            }
        }
    }

    private void handleError(SseEmitter emitter, ChatContext context, Exception e,Integer system) {
        log.error("Error while streaming data: {}", e.getMessage());

        if (context != null) {
            if (system == 1){
                contextRepository.save(context);
            } else if (system == 2){
                systemChatContextRepository.save(toSystem(context));
            }
        }

        synchronized (emitter) {
            try {
                if (e.getClass() == AuthException.class) {
                    emitter.send(SseEmitter.event().name("noAuth").data(e.getMessage()));
                } else {
                    emitter.send(SseEmitter.event().name("error").data(e.getMessage()));
                }
            } catch (IOException ioException) {
                log.error("Error while sending error event: {}", ioException.getMessage());
            } finally {
                log.info("handleError完成");
                emitter.complete();
            }
        }
    }

    private SystemChatContext toSystem(ChatContext context) {
        SystemChatContext systemChatContext = new SystemChatContext();
        systemChatContext.setId(context.getId());
        systemChatContext.setSessionId(context.getSessionId());
        systemChatContext.setRequestId(context.getRequestId());
        systemChatContext.setSessionTime(context.getSessionTime());
        systemChatContext.setSueContent(context.getSueContent());
        systemChatContext.setSysContent(context.getSysContent());
        systemChatContext.setFileIid(context.getFileIid());
        return systemChatContext;
    }
    private SystemChatSession toSystem(ChatSession chatSession,String systemName) {
        SystemChatSession systemChatSession = new SystemChatSession();
        systemChatSession.setId(chatSession.getId());
        systemChatSession.setSessionId(chatSession.getSessionId());
        systemChatSession.setSessionName(chatSession.getSessionName());
        systemChatSession.setRequestId(chatSession.getRequestId());
        systemChatSession.setUseId(chatSession.getUseId());
        systemChatSession.setSessionTime(chatSession.getSessionTime());
        systemChatSession.setFileId(chatSession.getFileId());
        systemChatSession.setDelState(chatSession.getDelState());
        systemChatSession.setSystemName(systemName);
        return systemChatSession;
    }


    private ChatContext buildChatContext(ChatModel model, String sessionIdWithTime, Date date) {
        ChatContext context = new ChatContext();
        context.setId(sessionIdWithTime);
        context.setRequestId(model.getRequestId());
        context.setSessionTime(date);
        context.setSueContent(model.getInput());
        context.setSessionId(model.getSessionId());
        context.setFileIid(String.join(",", model.getFileId()));
        return context;
    }

    private ChatSession buildChatSession(ChatModel model, String useId, String sessionIdWithTime, Date date) {
        ChatSession session = new ChatSession();
        session.setId(sessionIdWithTime);
        session.setRequestId(model.getRequestId());
        session.setSessionTime(date);
        session.setSessionName(model.getInput());
        session.setFileId(String.join(",", model.getFileId()));
        session.setUseId(useId);
        session.setDelState(0);
        session.setSessionId(model.getSessionId());
        return session;
    }


    @PostMapping(value = "/sse", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    public void doGet(HttpServletRequest request, HttpServletResponse response,@RequestBody ChatModel model)
            throws IOException {
        response.setContentType("text/event-stream");
        response.setCharacterEncoding("UTF-8");

        PrintWriter out = response.getWriter();

        try {
            ChatGPTClient client = new ChatGPTClient();
            ChatRequest chat = new ChatRequest(model);

            client.streamChat(chat, data -> {
                try {
                    ObjectMapper mapper = new ObjectMapper();
                    EleBillModel model1 = mapper.readValue(data, EleBillModel.class);
                    out.write(JSON.toJSONString(model1));
                    out.flush();

                } catch (IOException e) {
                    log.error("Error while processing data: {}", e.getMessage());
                    out.flush();
                }
            });
        } catch (Exception e) {
            log.error("Error while streaming data: {}", e.getMessage());
        }
        out.close();

    }

    //    @PostMapping(value = "/chatPost")
//    public void handleChatRequest(@RequestBody Map<String, Object> requestBody,
//                                  HttpServletRequest request,
//                                  HttpServletResponse response,@RequestBody ChatModel model) {
//        boolean stream = Boolean.parseBoolean(requestBody.getOrDefault("stream", "false").toString());
//        if (stream) {
//            String sessionId = request.getSession().getId();
//            log.info("sessionId: {}", sessionId);
//
//            PrintWriter writer = sessionEmitters.computeIfAbsent(sessionId, key -> {
//                try {
//                    return null;
//                } catch (IOException e) {
//                    log.error("Error while creating PrintWriter: {}", e.getMessage());
//                    throw new RuntimeException("Error while creating PrintWriter", e);
//                }
//            });
//
//            response.setContentType("text/event-stream");
//            response.setCharacterEncoding("UTF-8");
//
//            try {
//                ChatGPTClient client = new ChatGPTClient();
//                ChatRequest chat = new ChatRequest(model);
//
//                client.streamChat(chat, data -> {
//                    log.info("data是什么:{}",data);
//                    writer.println(data);
//                    writer.flush();
//                });
//            } catch (IOException e) {
//                log.error("Error while streaming data: {}", e.getMessage());
//                writer.println("event: error\n" + "data: " + e.getMessage());
//                writer.flush();
//            } finally {
//                closeWriter(writer, sessionId);
//            }
//        } else {
//            normalResponse(request, response);
//        }
//    }
    private void normalResponse(HttpServletRequest request, HttpServletResponse response) {
        try {
            String message = "这是普通响应的消息。";
            response.setCharacterEncoding("UTF-8");
            PrintWriter writer = response.getWriter();
            writer.println(message);
        } catch (IOException e) {
            log.error("Error while sending normal response: {}", e.getMessage());
        }
    }
    private void closeWriter(PrintWriter writer, String sessionId) {
        try {
            writer.println("event: end");
            writer.flush();
            writer.close();
        } catch (Exception ex) {
            log.error("Error closing PrintWriter for session {}: {}", sessionId, ex.getMessage());
        } finally {
            sessionEmitters.remove(sessionId);
        }
    }

    @Slf4j
    public static class ChatGPTClient {
        private final OkHttpClient client;

        private final String model = "gpt-3.5-turbo";

        public ChatGPTClient() {
            this.client = createOkHttpClient();
        }

        public static void main(String[] args) throws IOException {
            ChatRequest chat = new ChatRequest();
            chat.setConversation_id("11111111");
            chat.setDoc_content("[{\"role\": \"user\", \"content\": \"你好\"}, {\"role\": \"assistant\", \"content\": \"你好！很高兴能为你提供帮助。有什么问题或者需要讨论的话题吗？\"}, {\"role\": \"user\", \"content\": \"你能为我做些什么\"}, {\"role\": \"assistant\", \"content\": \"我可以回答你的问题、提供定义和解释、将文本从一种语言翻译成另一种语言、总结文本、生成文本、编写故事、分析情感、提供推荐、开发算法、编写代码以及其他任何基于语言的任务。\"}]");
            chat.setInput("请分析这份电费账单");
            new ChatGPTClient().streamChat(chat, data -> {
                log.info("data:{}", data);
            });
        }

        private OkHttpClient createOkHttpClient() {
            return new OkHttpClient.Builder()
                    .readTimeout(60, TimeUnit.SECONDS)
                    .build();
        }

        public String streamChat(String message) {
            return new HttpUtil().okHttpPost(
                    "https://serverfastgpt.excn.top/api/v1/chat/completions"
                    , "{\"model\": \"" + model + "\", \"messages\": [{\"role\": \"user\", \"content\": \"" + message + "\"}], \"stream\": true}");

        }

        public void streamChat(ChatRequest chatRequest, StreamListener listener) {
            okhttp3.MediaType mediaType = okhttp3.MediaType.get("application/json");
            okhttp3.RequestBody body = okhttp3.RequestBody.create(mediaType, JSON.toJSONString(chatRequest));
            Request request = new Request.Builder()
                    .url("http://112.111.7.68:13201/ele_bill")
                    .header("Authorization", "Basic YmFja2VuZF9hY2NvdW50OmJhY2tlbmRfMjAyNDA1MzA=")
                    .addHeader("Accept", "text/event-stream")
                    //                .addHeader("Content-Type", "application/json; charset=UTF-8")
                    .post(body)
                    .build();
            new Thread(() -> {
                try (Response response = client.newCall(request).execute()) {
                    if (!response.isSuccessful())
                        throw new IOException("Request failed with HTTP error code: " + response.code());

                    ResponseBody responseBody = response.body();
                    if (responseBody == null) return;

                    try (BufferedSource source = responseBody.source()) {
                        while (!source.exhausted()) {
                            String data = source.readUtf8Line();
                            if (data != null) {
                                listener.onData(data);
                            }
                        }
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }).start();
        }


        interface StreamListener {
            void onData(String data) throws JsonProcessingException;
        }
    }
}
